package com.itdct.mto.core.manager.base;

import com.itdct.mto.core.base.BaseManager;
import com.itdct.mto.core.bo.ColumnBo;
import com.itdct.mto.core.bo.TableBo;
import com.itdct.mto.core.context.ColumnContext;
import com.itdct.mto.core.context.Context;
import com.itdct.mto.core.context.TableContext;

import org.springframework.beans.factory.annotation.Autowired;

import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.extern.slf4j.Slf4j;

/**
 * @author Zhouwx
 * @date 2025/7/7 10:15:50
 * @version 1.0
 * @description
 */
@Slf4j
public abstract class BaseDataTransferManager extends BaseManager {
    @Autowired
    private ThreadPoolExecutor dataTransferExecutor;

    public void transferData(Context context) {
        log.info("开始将MySQL中的数据传输到OpenGauss {}模式", mtoConfig.getDatabaseMode());
        init(context);
        List<TableBo> tableBos = context.getTableBos();
        List<String> failTableNames = context.getFailTableNames();
        context.setTransferStartTime(System.currentTimeMillis());

        CountDownLatch countDownLatch = new CountDownLatch(tableBos.size());

        for (TableBo tableBo : tableBos) {
            String tableName = tableBo.getTableName();
            if (tableBo.getException() != null) {
                countDownLatch.countDown();
                continue;
            }

            int waitCount = 0;
            BlockingQueue<Runnable> queue = threadPoolExecutor.getQueue();
            while (true) {
                if (queue.size() >= mtoConfig.getThreadCount()) {
                    // INFO: Zhouwx: 2025/7/2 等待队列过长，卡住整个循环，防止内存溢出
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        log.error("", e);
                    }
                    waitCount++;
                    if (waitCount % 10 == 0) {
                        log.info("等待队列过长，已等待 {} 秒", (waitCount * 200) / 1000);
                    }
                } else {
                    break;
                }
            }

            // INFO: Zhouwx: 2025/7/3 提交数据迁移任务到线程池
            threadPoolExecutor.submit(() -> {
                try {
                    tableBo.setTransferStartTime(System.currentTimeMillis());
                    doTransferData(tableBo, context);
                    tableBo.setTransferSuccess(true);
                } catch (Exception e) {
                    log.error("表名：{}，执行失败", tableName);
                    log.error("", e);
                    failTableNames.add(tableName);
                    tableBo.setException(e);
                    tableBo.setTransferSuccess(false);
                } finally {
                    tableBo.setTransferEndTime(System.currentTimeMillis());
                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("", e);
        }
        log.info("完成将MySQL中的数据传输到OpenGauss中，总共：{} 张表，失败 {} 张表", tableBos.size(), failTableNames.size());
        for (String failTable : failTableNames) {
            log.error("表名：{}，执行失败", failTable);
        }
        context.setTransferEndTime(System.currentTimeMillis());
    }

    protected void doTransferData(TableBo tableBo, Context context) {
        String tableName = tableBo.getTableName();
        int batchSize = mtoConfig.getBatchSize();
        TableContext tableContext = new TableContext();
        long startTime = System.currentTimeMillis();
        mysqlJdbcTemplate.query("select count(0) from " + tableName, rs -> {
            long count = rs.getLong(1);
            tableContext.setTotalCount(count);
            tableBo.setTotalCount(count);
        });

        if (tableContext.getTotalCount() == 0) {
            log.info("表名：{}，没有数据，跳过", tableName);
            return;
        }

        if (mtoConfig.isTruncateTable()) {
            postgresJdbcTemplate.execute("truncate table " + tableName);
        }
        log.info("开始将表名：{} 的数据进行转换", tableName);

        // INFO: Zhouwx: 2025/7/3 处理主键信息，用于滚动式查询，提高查询效率
        String primaryKey = handlePrimaryKey(tableBo, tableContext);

        ArrayList<ColumnBo> columnList = tableBo.getColumnList();
        int totalPage = (int) Math.ceil((double) tableContext.getTotalCount() / batchSize);
        tableContext.setTotalPage(totalPage);
        long lastMonitorTime = System.currentTimeMillis();
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);
        for (int pageNumber = 1; pageNumber <= totalPage; pageNumber++) {
            // INFO: Zhouwx: 2025/7/2 进行分页查询，防止内存溢出
            String selectSql = generateSelectSql(primaryKey, pageNumber, tableName, batchSize, tableContext);

            StringBuilder insertBuilder = new StringBuilder();
            insertBuilder.append("INSERT INTO ")
                    .append(tableName)
                    .append(" VALUES ");
            final int[] count = {0};
            String finalPrimaryKey = primaryKey;
            mysqlJdbcTemplate.query(selectSql, (rs) -> {
                insertBuilder.append("(");
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                if (columnCount == 0) {
                    log.warn("表名：{}，无字段", tableName);
                    return;
                }
                for (int columnIndex = 1; columnIndex <= columnCount; columnIndex++) {
                    String columnName = metaData.getColumnName(columnIndex);
                    final AtomicBoolean columnNotNull = new AtomicBoolean(false);
                    columnList.stream()
                            .filter(columnBo -> columnBo.getColumnName().equals(columnName))
                            .findFirst()
                            .ifPresent(columnBo -> {
                                columnNotNull.set(columnBo.isNotNull());
                            });
                    String columnType = metaData.getColumnTypeName(columnIndex).toUpperCase(Locale.ROOT);
                    String columnValue = rs.getString(columnName);

                    ColumnContext columnContext = new ColumnContext();
                    columnContext.setColumnName(columnName);
                    columnContext.setColumnType(columnType);
                    columnContext.setNotNull(columnNotNull.get());
                    columnContext.setColumnValue(columnValue);
                    columnContext.setRs(rs);
                    columnContext.setTableBo(tableBo);
                    columnContext.setTableContext(tableContext);
                    columnContext.setMetaData(metaData);
                    try {
                        appendInsertLineSql(insertBuilder, columnContext);
                    } catch (Exception e) {
                        log.error("", e);
                        throw new RuntimeException(e);
                    }

                    if (columnIndex < columnCount) {
                        insertBuilder.append(",");
                    }

                    if (columnName.equals(finalPrimaryKey)) {
                        tableContext.setLastPrimaryKeyValue(columnValue);
                    }
                }
                insertBuilder.append("),");
                count[0]++;
            });
            tableContext.setTransferCount(tableContext.getTransferCount() + count[0]);

            if (count[0] == 0) {
                // INFO: Zhouwx: 2025/7/7 数据库中已经没有数据了，全部倒数完，然后跳出循环
                while (countDownLatch.getCount() > 0) {
                    countDownLatch.countDown();
                }
                break;
            }

            insertBuilder.deleteCharAt(insertBuilder.length() - 1);

            // INFO: Zhouwx: 2025/7/7 执行到目标数据库单独使用新的线程池，提高执行效率
            BlockingQueue<Runnable> queue = dataTransferExecutor.getQueue();
            while (true) {
                if (queue.size() > mtoConfig.getThreadCount()) {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        log.error("", e);
                    }
                } else {
                    break;
                }
            }

            dataTransferExecutor.submit(() -> {
                String insertSql = insertBuilder.toString();
                try {
                    postgresJdbcTemplate.execute(insertSql);
                } catch (Exception e) {
                    log.error("", e);
                    context.getFailTableNames().add(tableName);
                    tableBo.setException(e);
                    tableBo.setTransferSuccess(false);
                } finally {
                    countDownLatch.countDown();
                }
            });

            if (primaryKey == null) {
                // INFO: Zhouwx: 2025/7/2 无主键的情况要改offset
                tableContext.setOffset(tableContext.getOffset() + batchSize);
            }

            // INFO: Zhouwx: 2025/7/3 输出监控信息
            if (System.currentTimeMillis() - lastMonitorTime > mtoConfig.getTransferMonitorInterval()) {
                log.info("表名：{}，传输数量 {} 条数据，耗时：{}秒", tableName, tableContext.getTransferCount() + " / " + tableContext.getTotalCount(), (System.currentTimeMillis() - startTime) / 1000);
                lastMonitorTime = System.currentTimeMillis();
            }
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("", e);
        }
        long endTime = System.currentTimeMillis();
        log.info("表名：{}，共 {} 条数据，分 {} 页，数据转换完成，耗时：{}秒", tableName, tableContext.getTotalCount(), totalPage, (endTime - startTime) / 1000);
    }

    protected abstract void appendInsertLineSql(StringBuilder insertBuilder, ColumnContext columnContext) throws Exception;

    protected String generateSelectSql(String primaryKey, int pageNumber, String tableName, int batchSize, TableContext tableContext) {
        StringBuilder stringBuilder = new StringBuilder();
        String selectSql = null;
        if (primaryKey != null) {
            // INFO: Zhouwx: 2025/7/2 存在主键，采用id推进的方式提高查询效率
            if (pageNumber == 1) {
                // INFO: Zhouwx: 2025/7/2 第一页的时候不需要primaryKey大于多少，防止uuid组成的id无法获取
                stringBuilder.append("SELECT * FROM ")
                        .append(tableName)
                        .append(" ORDER BY ")
                        .append(primaryKey)
                        .append(" ASC ")
                        .append("limit ")
                        .append(batchSize);
                selectSql = stringBuilder.toString();
            } else {
                // INFO: Zhouwx: 2025/7/2 后面几页为了提高查询效率，需要大于最后一个primaryKey
                stringBuilder.append("SELECT * FROM ")
                        .append(tableName)
                        .append(" where ")
                        .append(primaryKey)
                        .append(" > '")
                        .append(tableContext.getLastPrimaryKeyValue())
                        .append("' ORDER BY ")
                        .append(primaryKey)
                        .append(" ASC ")
                        .append("limit ")
                        .append(batchSize);
                selectSql = stringBuilder.toString();
            }
        } else {
            // INFO: Zhouwx: 2025/7/2 一张表不存在主键，那就毁灭吧！效率太低怪我咯？
            stringBuilder.append("SELECT * FROM ")
                    .append(tableName)
                    .append(" limit ")
                    .append(batchSize)
                    .append(" offset ")
                    .append(tableContext.getOffset());
            selectSql = stringBuilder.toString();
        }
        return selectSql;
    }

    protected String handlePrimaryKey(TableBo tableBo, TableContext tableContext) {
        tableBo.getIndexList()
                .stream()
                .filter(indexBo -> indexBo.getIndexType().equals("PRIMARY KEY"))
                .findFirst()
                .ifPresent(indexBo -> {
                    tableContext.setPrimaryKey(indexBo.getColumns());
                });

        String primaryKey = tableContext.getPrimaryKey();
        if (primaryKey != null && primaryKey.contains(",")) {
            // INFO: Zhouwx: 2025/7/3 如果主键存在多个字段，那就取第一个字段
            primaryKey = primaryKey.split(",")[0];
        }
        return primaryKey;
    }
}
